package com.run.wz.test;


import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTableInterfaceFactory;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import com.google.protobuf.Descriptors.FieldDescriptor;
import com.run.pbf.Object.ObjectRelation;

public class SecondaryIndex {
	private static Configuration config;
	private static HTable table;
	private static byte[][] nums = {{(byte)49},{(byte)50},{(byte)51},{(byte)52},{(byte)53},{(byte)54},{(byte)55},{(byte)56},{(byte)57}};
	private static ExecutorService pool = new ThreadPoolExecutor(100,100,60,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(100));
	private static HTablePool htablePool;
	
	private static void init() throws IOException {
		config = new Configuration();
		config.set("hbase.zookeeper.quorum", "DG1HB3,DG1HB4,DG1HB5");
		config.set("hbase.client.ipc.pool.size", "8");
		config.set("hbase.zookeeper.property.clientPort", "2181");
		table = new HTable(config, "object_relation_201304");
		htablePool = new HTablePool(config, 100, new HTableInterfaceFactory() {
			
			@Override
			public void releaseHTableInterface(HTableInterface table)
					throws IOException {
				table.close();
			}
			
			@Override
			public HTableInterface createHTableInterface(Configuration config,
					byte[] tableName) {
				HTable hTable = null;
				try {
					hTable = new HTable(tableName,HConnectionManager.getConnection(config),pool);
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				hTable.setAutoFlush(false);
				try {
					hTable.setWriteBufferSize(4*1024*1024);
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				return hTable;
			}
		});
		for(int i=0; i<100; i++) {
			HTableInterface table2 = htablePool.getTable("chenxu_secondary_index_1");
			table2.close();
		}
	}
	
	private static void trans() throws IOException, InterruptedException, ExecutionException {
		Map<Integer,List<Put>> putsMap = new HashMap<Integer, List<Put>>();
		long count = 0;
		Scan scan = new Scan();
		int caching = 1000;
		scan.setCaching(caching);
		scan.setCacheBlocks(true);
		List<Put> puts = new ArrayList<Put>();
		Random r = new Random();
		HTable indexT = null;
		while(count<200000) {
			ResultScanner scanner = table.getScanner(scan );
			Result[] next = scanner.next(caching);
			count+=caching; long s = System.currentTimeMillis();
			try {
				for (int i = 0; i < next.length; i++) {
					KeyValue[] raw = next[i].raw();
					for (int j = 0; j < raw.length; j++) {
						KeyValue keyValue = raw[j];
						ObjectRelation parseFrom = ObjectRelation.parseFrom(keyValue.getValue());
						Map<FieldDescriptor, Object> allFields = parseFrom.getAllFields();
						for(Entry<FieldDescriptor, Object> entry : allFields.entrySet()) {
							FieldDescriptor key = entry.getKey();
							Object value = entry.getValue();
							
//						System.out.println(key.getNumber());
//						byte[] add = Bytes.add(Bytes.add(Bytes.toBytes(key.getNumber()), keyValue.getQualifier()), Bytes.add(nums[r.nextInt(9)], nums[r.nextInt(9)]));
							byte[] qualifier = keyValue.getQualifier();
//						System.out.println(Arrays.toString(qualifier));
							byte[] add = Bytes.add(Bytes.add(nums[key.getNumber()], qualifier), Bytes.add(nums[r.nextInt(9)], nums[r.nextInt(9)]));
//						System.out.println(Arrays.toString(add));
							Put put = new Put(add);
							
//						System.out.println(Bytes.toLong(Bytes.add(Bytes.toBytes(key.getNumber()), Bytes.toBytes(currentTimeMillis))));
							
							put.add(Bytes.toBytes("r"), Bytes.toBytes(value.toString()),System.currentTimeMillis() , parseFrom.toByteArray());
							puts.add(put);
						}
					}
				}
				indexT = (HTable) htablePool.getTable("chenxu_secondary_index_1");
				indexT.put(puts);
				indexT.flushCommits();
			} catch (Exception e) {
				
			} finally {
				puts.clear();
				indexT.close();
			}
			
			System.out.println("count:"+count+" all_time: "+(System.currentTimeMillis()-s));
		}
		
	}
	
	public static void main(String[] args) throws IOException, InterruptedException, ExecutionException {
		init();
		trans();
		pool.shutdown();
//	System.out.println(Arrays.toString(Bytes.toBytes("12")));	
	}

}
